package com.leilei;


import cn.hutool.core.util.RandomUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Random;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/19 20:41
 * @desc chekpoint
 * 一般情况下，如果对我们的计算逻辑没有try-catch,一旦抛出了异常，整个计算服务便会停止，
 * 且之前我们所有的状态都会丢失，但是使用checkpoint,我们可以控制计算程序是否重启以及控制我们的状态存储位置，
 * 当计算程序重启后，会沿用之前的状态继续计算
 * https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/state/checkpointing.html
 * <p>
 * check-point 重启策略设置
 */
public class Flink_CheckPoint_2_Restart {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(4);
        //todo 下边为check-point设置
        //设置Checkpoint的时间间隔为1000ms做一次保存发一次Barrier（保存状态state）!
        env.enableCheckpointing(1000);
        // 状态存储保存位置
        env.setStateBackend(new FsStateBackend("file:///usr/local/over-speeding"));
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置多个checkpoint之间执行最小间隔时间500ms 。此设置后 setMaxConcurrentCheckpoints 无效
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //如果checkpoint 保存超时时间，超过60000ms未成功，则丢弃此次操作
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 设置checkpoint并行度 （一次又几个执行 ）
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // RETAIN_ON_CANCELLATION启用在取消工作后保留的外部检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        //checkpoint操作失败次数限制，默认为0，即不允许checkpoint失败（一旦失败，整个计算程序终止）
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(5);
        //todo check-point 设置后，未指定重启策略，代码中出现了非致命错误时（oom之类的）,程序会无限重启，也不会打印出异常堆栈信息
        //设置重启策略
        //默认：无限重启
        //设置：禁止重启，项目出现bug，但程序未进行捕获，计算程序会直接挂掉
        //  env.setRestartStrategy(RestartStrategies.noRestart());
        //设置：延迟重启    最多重启20次，每次间隔十秒钟，比如程序刚启动好，又出现bug,程序延迟十秒启动
         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(20, Time.seconds(10)));
        // 失败率重启：如果5分钟内job失败不超过三次,自动重启, 每次间隔10s (如果5分钟内程序失败超过3次,则程序退出）
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, // 每个测量时间间隔最大失败次数
//                Time.minutes(5), //失败率测量的时间间隔
//                Time.seconds(10) // 每次重启的时间间隔
//        ));
        //加载数据源
        final DataStreamSource<Location> locationSource = env.addSource(new LocationSource());
        //新开窗口
        final WindowedStream<Location, Integer, GlobalWindow> windowedStream = locationSource
                .keyBy(Location::getVehicleId)
                .countWindow(2);
        //执行窗口计算逻辑
        DataStream<Tuple2<String, Integer>> result = windowedStream.apply(new SpeedAlarmWindow());
        result.print();
        String sql = "insert into state_history (`key`,`count`) values(?,?)";
        JdbcConnectionOptions jdbcBuild = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName("com.mysql.jdbc.Driver")
                .withUrl("jdbc:mysql://xxx:3306/flink?useUnicode=true&characterEncoding=utf-8&useSSL=false")
                .withUsername("root")
                .withPassword("root")
                .build();

        result.addSink(JdbcSink.sink(sql, (ps, tuple2) -> {
            ps.setString(1, tuple2.f0);
            ps.setInt(2, tuple2.f1);
        }, jdbcBuild));
        env.execute("check-point-restart");

    }


    public static class SpeedAlarmWindow extends RichWindowFunction<Location, Tuple2<String, Integer>, Integer, GlobalWindow> {
        MapState<String, Integer> locationState;

        @Override
        public void apply(Integer integer, GlobalWindow window, Iterable<Location> locationList, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (Location location : locationList) {
                final String key = location.getVehicleId().toString();
                Integer overSpeedCount = locationState.get(key);
                if (location.getGpsSpeed() > location.getLimitSpeed()) {
                    Integer count = overSpeedCount == null ? 1 : overSpeedCount + 1;
                    locationState.put(key, count);
                    out.collect(Tuple2.of(location.getVehicleId() + "-" + location.getPlate() + location.getColor(), count));
                    if (location.getVehicleId() == 2) {
                        //由于设置了checkpoint,即使2抛出了异常，但1的状态会一直累计
                        System.out.println("bug........");
                        throw new Exception("bug........");
                    }
                }
            }
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            locationState = getRuntimeContext().getMapState(new MapStateDescriptor<>("locationState",
                    TypeInformation.of(String.class),
                    TypeInformation.of(Integer.class)));

        }

    }

    public static class LocationSource implements SourceFunction<Location> {

        Boolean flag = true;

        @Override
        public void run(SourceContext<Location> ctx) throws Exception {
            Random random = new Random();
            while (flag) {
                int vehicleId = random.nextInt(2) + 1;
                Location location = Location.builder()
                        .vehicleId(vehicleId)
                        .plate("川A000" + vehicleId)
                        .color("绿")
                        .date(Integer.parseInt(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)))
                        .gpsSpeed(RandomUtil.randomInt(88, 100))
                        .limitSpeed(RandomUtil.randomInt(88, 95))
                        .devTime(System.currentTimeMillis())
                        .build();
                ctx.collect(location);
                Thread.sleep(3000);

            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}

